package yuekao8.util;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.java.tuple.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import yuekao8.entity.*;

import java.sql.*;
import java.sql.DriverManager;
import java.util.Collections;

public class AsyncIOUtil extends RichAsyncFunction<Tuple2<OrderInfo, OrderDetail>, Tuple6<OrderInfo, OrderDetail, ProductSpu, ProductSku, Region, Shop>> {
    Connection conn = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181", "", "");
        conn.setAutoCommit(true);
    }

    @Override
    public void close() throws Exception {
        conn.close();
    }

    @Override
    public void asyncInvoke(Tuple2<OrderInfo, OrderDetail> value, ResultFuture<Tuple6<OrderInfo, OrderDetail, ProductSpu, ProductSku, Region, Shop>> resultFuture) throws Exception {
        ProductSpu productspu = null;
        ProductSku productSku = null;
        // 先用orderdetail关联product_sku表给此表赋值,然后在用赋完值的product_sku取出Product_spu_id关联dim_product_spu表
        try (PreparedStatement pds = conn.prepareStatement("select * from dim_product_sku where id=?")) {
            pds.setObject(1, value.f1.getProduct_sku_id());
            ResultSet rs = pds.executeQuery();
            if (rs.next()) {
                String line = rs.getString("line");
                String data = JSON.parseObject(line).getString("data");
                productSku = JSON.parseObject(data, ProductSku.class);

                try (PreparedStatement pds1 = conn.prepareStatement("select * from dim_product_spu where id=?")) {
                    pds1.setObject(1, productSku.getProduct_spu_id());
                    ResultSet rs1 = pds1.executeQuery();
                    if (rs1.next()) {
                        String line1 = rs1.getString("line");
                        String data1 = JSON.parseObject(line1).getString("data");
                        productspu = JSON.parseObject(data1, ProductSpu.class);
                        rs.close();
                        rs1.close();
                    }
                }
            }
        }

        Region region = null;
        Shop shop = null;

        try (PreparedStatement pds = conn.prepareStatement("select * from dim_shop where id=?")) {
            pds.setObject(1, value.f0.getShop_id());
            ResultSet rs = pds.executeQuery();
            if (rs.next()) {
                String line = rs.getString("line");
                String data = JSON.parseObject(line).getString("data");
                shop = JSON.parseObject(data, Shop.class);
                try (PreparedStatement pds1 = conn.prepareStatement("select * from dim_region where id=?")) {
                    pds1.setObject(1, shop.getRegion_id());
                    ResultSet rs1 = pds1.executeQuery();
                    if (rs1.next()) {
                        String line1 = rs1.getString("line");
                        String data1 = JSON.parseObject(line1).getString("data");
                        region = JSON.parseObject(data1, Region.class);
                        rs.close();
                        rs1.close();
                    }
                }
            }
        }

        resultFuture.complete(Collections.singleton((new Tuple6<>(value.f0, value.f1, productspu, productSku, region, shop))));
    }

}
